package com.cskaoyan.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;

/**
 * @author LY
 * @since 2022/06/27 21:41
 */
@Slf4j
@Component
public class PromoOrderProducer {

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;

    @Autowired
    RedissonClient redissonClient;

    private TransactionMQProducer producer;

    @PostConstruct
    public void init() {
        producer = new TransactionMQProducer("promo_order_producer");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                Map<String, Long> paramMap = (Map<String, Long>) o;
                Long productId = paramMap.get("productId");
                Long psId = paramMap.get("psId");
                String localTransactionKey = "promo_local_transaction: " + message.getTransactionId();

                String lockKey = "promo_stock_lock " + productId + ":" + psId;
                Lock lockObj = redissonClient.getLock(lockKey);
                Integer affectedRows = 0;

                lockObj.lock();
                try {
                    affectedRows = promoItemMapper.decreaseStock(productId, psId);
                } finally {
                    lockObj.unlock();
                }
                String promoItemStockFlag = "promo_item_stock_not_enough:" + productId + ":" + psId;
                if (affectedRows != 1) {
                    cacheManager.setCache(promoItemStockFlag, "none", 1);
                    cacheManager.setCache(localTransactionKey, "fail", 1);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                cacheManager.setCache(localTransactionKey, "success", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                String localTransactionKey = "promo_local_transaction: " + messageExt.getTransactionId();
                String state = cacheManager.checkCache(localTransactionKey);
                if (state == null || state.trim().isEmpty()) {
                    return LocalTransactionState.UNKNOW;
                } else if ("success".equals(state)) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public boolean sendPromoOrderMessage(CreateSeckillOrderRequest createSeckillOrderRequest, Long productId, Long psId) {
        String requestJson = JSON.toJSONString(createSeckillOrderRequest);

        Message message = new Message();
        message.setTopic("promo_order");
        message.setBody(requestJson.getBytes(Charset.forName("utf-8")));
        Map<String, Long> arg = new HashMap<>();
        arg.put("productId", productId);
        arg.put("psId", psId);
        TransactionSendResult result = null;
        try {
            result = producer.sendMessageInTransaction(message, arg);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        if (result != null && LocalTransactionState.COMMIT_MESSAGE.equals(result.getLocalTransactionState())) {
            return true;
        }
        return false;
    }
}
